package cn.boat.router.consumer;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;

public class SubscribeTopcisImpl implements SubscribeTopcisInterface {
    private static final String BROKERS = "/boatMQ/brokers";
    private static final String TOPICS = "/topics";
    private static final String PARTITIONS = "/partitions";
    private static final String CONSUMERS = "/boatMQ/consumers";
    private static final String IDS = "/ids";
    private static final String OWNERS = "/owners";
    private static final String STATE = "/state";

    /*private int consumerGroupSize;
    private int subTopicsPartionSize;*/
    ZooKeeper zooKeeperConnection;
/*    public SubscribeTopcisReal(List<String> topics) {
        this.topics = topics;
        zooKeeperConnection=null;
    }
    public SubscribeTopcisReal(String topic) {
        zooKeeperConnection=null;
        this.topic = topic;
    }*/

    public SubscribeTopcisImpl() {
        zooKeeperConnection=null;
    }



    @Override
    public ConsumerMetaData subscribeTopics(ZooKeeper zooKeeperConnection, String consumerGroupId, String consumerID, String topic) throws Exception {
        List<String> ifTopicList = zooKeeperConnection.getChildren(BROKERS + TOPICS , false);
        Collections.sort(ifTopicList);
        boolean flag =false;
        for (String ifTopic : ifTopicList) {
            if (ifTopic == topic){
                flag =true;
                break;
            }
        }
        if (!flag){
            System.out.println("您订阅的topic不存在，请重新更换topic");
            return null;
        }
        int consumerGroupSize = 0;
        int subTopicsPartionSize  =0;
        List<String> consumerIdList = new ArrayList<>();
        // 判断消费者组的消费者是否订阅过该topic
        Stat stat = zooKeeperConnection.exists(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic, false);

        // 消费者组的消费者订阅过该topic，返回HashMap<String topic,List<TopicPartitionBrokerLeader>>
        if (stat!=null){
            System.out.println("你是："+consumerID+"该"+topic+"已经被"+consumerGroupId+"同一组的成员订阅过，请不要重复订阅");
            GetConsumerMetaDataImpl getConsumerMetaDataMy =new GetConsumerMetaDataImpl();
            ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerID, zooKeeperConnection);
            return consumerMetaData;
        }

        // 消费者组的消费者没有订阅过该topic,返回HashMap<String topic,List<TopicPartitionBrokerLeader>>
        MyLock myLock =new MyLock(zooKeeperConnection);
        myLock.acquireLock();
        zooKeeperConnection.create(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic,topic.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        myLock.releaseLock();
        List<String> partitionsList = zooKeeperConnection.getChildren(BROKERS + TOPICS + "/" + topic + PARTITIONS, false);
            Collections.sort(partitionsList);
            if (partitionsList.size()!=0){
                subTopicsPartionSize = partitionsList.size();
                for (String partitionNum : partitionsList) {
                    zooKeeperConnection.create(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionNum,partitionNum.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
            }

            List<String> consumerList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroupId + IDS, false);
            Collections.sort(consumerList);
            if (consumerList.size()!=0){
                consumerGroupSize = consumerList.size();
                for (String myConsumerId : consumerList) {
                    System.out.println(myConsumerId);
                    byte[] data = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroupId + IDS + "/" + myConsumerId, false, null);
                    System.out.println("数据为"+new String(data));
                    String dataStr = new String(data);
                    consumerIdList.add(dataStr); // 将consumer id加入到List集合中，以便做负载均衡
                }
            }


            // 负载均衡：
            if (consumerGroupSize!=0 && subTopicsPartionSize!=0){

                if (consumerGroupSize>subTopicsPartionSize){

                        for (int i = 0; i < subTopicsPartionSize; i++) {
                            zooKeeperConnection.setData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionsList.get(i),(consumerIdList.get(i)).getBytes(),-1);
                            /*TopicPartitionConsumerId topicPartitionConsumerId = new TopicPartitionConsumerId();
                            topicPartitionConsumerId.set(partitionsList.get(i),consumerIdList.get(i));
                            topicPartitionConsumerIdList.add(topicPartitionConsumerId);*/

                        }


                }

                if (consumerGroupSize==subTopicsPartionSize){
                    for (int i = 0; i < subTopicsPartionSize; i++) {
                        zooKeeperConnection.setData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionsList.get(i),(consumerIdList.get(i).getBytes()),-1);
                    }
                }
                if (consumerGroupSize<subTopicsPartionSize){
                    int yuShu = subTopicsPartionSize%consumerGroupSize;
                    int averageNum = subTopicsPartionSize / consumerGroupSize;
                    /*for (String s : consumerIdList) {
                        System.out.println(s);
                    }*/
                        for (int i = 0, j=0; i < subTopicsPartionSize; i++,j++) {
                            if (i>0 && ((i+consumerGroupSize) % consumerGroupSize) ==0){
                                averageNum--;
                                j=0;
                            }
                            if (averageNum>0){
                                System.out.println(consumerIdList.get(j));
                                zooKeeperConnection.setData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionsList.get(i),((consumerIdList.get(j)).getBytes()),-1);
                            }
                            if (averageNum ==0){
                                if (yuShu>0){
                                    zooKeeperConnection.setData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionsList.get(i),(consumerIdList.get(j).getBytes()),-1);
                                    yuShu--;
                                }
                            }
                        }

                }
            }
        GetConsumerMetaDataImpl getConsumerMetaDataMy =new GetConsumerMetaDataImpl();
        ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerID, zooKeeperConnection);
        return consumerMetaData;
    }

    @Override
    public ConsumerMetaData subscribeTopics(ZooKeeper zooKeeperConnection, String consumerGroupId, String consumerID, String topic, int partitionQuantity) throws Exception {
        List<String> ifTopicList = zooKeeperConnection.getChildren(BROKERS + TOPICS , false);
        Collections.sort(ifTopicList);
        boolean flag =false;
        for (String ifTopic : ifTopicList) {
            if (ifTopic == topic){
                flag =true;
                break;
            }
        }
        if (!flag){
            System.out.println("您订阅的topic不存在，请重新更换topic");
            return null;
        }
        List<String> partitionsList = zooKeeperConnection.getChildren(BROKERS + TOPICS + "/" + topic + PARTITIONS, false);
        Collections.sort(partitionsList);
        /*List<TopicPartitionBrokerLeader> topicPartitionBrokerLeaderList = new ArrayList<>();
        HashMap<String,List<TopicPartitionBrokerLeader>>  topicPartitionBrokerLeaderMap= new HashMap<>();*/
        /*int consumerGroupSize = 0;
        int subTopicsPartionSize  =0;
        List<String> consumerIdList = new ArrayList<>();*/
        // 判断消费者组的消费者是否订阅过该topic
        Stat stat = zooKeeperConnection.exists(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic, false);

        // 消费者组的消费者订阅过该topic，返回HashMap<String topic,List<TopicPartitionBrokerLeader>>
        if (stat!=null){
            System.out.println("你是："+consumerID+"该"+topic+"已经被"+consumerGroupId+"同一组的成员订阅过,现在给你分配分区");

            int alreadySubTopicCount = 0;
            List<String> subTopicPartitionList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic, false);
            Collections.sort(subTopicPartitionList);
            for (String subTopicPartitionNum : subTopicPartitionList) {
                byte[] consumerIdTemp = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic + "/" + subTopicPartitionNum, false, null);
                if (!Objects.equals(new String(consumerIdTemp),"null404")){
                    alreadySubTopicCount++;
                }
            }
            if (partitionQuantity<0){
                System.out.println("订阅分区数量不能小于0");
                return null;
            }
            int canSubTopicCount = partitionsList.size()-alreadySubTopicCount; // 可以订阅分区数量
            if (canSubTopicCount<partitionQuantity){
                System.out.println("对不起，你订阅的主题分区数量超过剩余分区数量，订阅分区数量不要超过"+canSubTopicCount+"个");
                return null;
            }
            int canConsumerTopicConut = partitionQuantity;
            for (String partitionNum : partitionsList) {
                byte[] consumerPartition = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic + "/" + partitionNum, false, null);
                if (canConsumerTopicConut>0){
                    if (Objects.equals("null404",new String(consumerPartition)) ){
                        zooKeeperConnection.setData(CONSUMERS+"/"+consumerGroupId+OWNERS+"/"+topic+"/"+partitionNum,consumerID.getBytes(),-1);
                        canConsumerTopicConut--;
                    }
                }
            }
            GetConsumerMetaDataImpl getConsumerMetaDataMy =new GetConsumerMetaDataImpl();
            ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerID, zooKeeperConnection);
            return consumerMetaData;
        }



        // 消费者组的消费者没有订阅过该topic,返回HashMap<String topic,List<TopicPartitionBrokerLeader>>

        zooKeeperConnection.create(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic,topic.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        if (partitionsList.size()!=0){
            for (String partitionNum : partitionsList) {
                // 创建消费者owenrs节点下订阅过的主题节点下的分区，第一个消费者都先创建好，默认值都为"null404",如果该分区分配给某个消费者，存放该消费者Id
                zooKeeperConnection.create(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic+"/"+partitionNum,"null404".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
        int alreadySubTopicCount = 0;
        List<String> subTopicPartitionList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic, false);
        Collections.sort(subTopicPartitionList);
        for (String subTopicPartitionNum : subTopicPartitionList) {
            byte[] consumerIdTemp = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic + "/" + subTopicPartitionNum, false, null);
            if (!Objects.equals(new String(consumerIdTemp),"null404")){
                alreadySubTopicCount++;
            }
        }
        if (partitionQuantity<0){
            System.out.println("订阅分区数量不能小于0");
            return null;
        }
        int canSubTopicCount = partitionsList.size()-alreadySubTopicCount; // 可以订阅分区数量
        if (canSubTopicCount<partitionQuantity){
            System.out.println("对不起，你订阅的主题分区数量超过剩余分区数量，订阅分区数量不要超过"+canSubTopicCount+"个");
            return null;
        }
        int canConsumerTopicConut = partitionQuantity;
        for (String partitionNum : partitionsList) {
            if (canConsumerTopicConut>0){
                zooKeeperConnection.setData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + topic + "/"+partitionNum,consumerID.getBytes(),-1);
                canConsumerTopicConut--;
            }
        }
        GetConsumerMetaDataImpl getConsumerMetaDataMy =new GetConsumerMetaDataImpl();
        ConsumerMetaData consumerMetaData = getConsumerMetaDataMy.getConsumerMetaData(consumerGroupId, consumerID, zooKeeperConnection);
        return consumerMetaData;
    }
}
